package com.xj.zookeeper.framework;

import org.apache.log4j.Logger;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;

import java.util.List;
import java.util.concurrent.Semaphore;

/**
 * Author: xiajun
 * Date: 2014-08-07
 * Time: 16:05:00
 * 事件通知处理
 */
public class ZkWatcher implements Watcher, AsyncCallback.ChildrenCallback {
    private final static Logger log = Logger.getLogger(ZkWatcher.class);
    private Semaphore connectionLock;
    private boolean asnycConnection;
    private WatcherProcess process;
    private ZkClient zkClient;

    public ZkWatcher(Semaphore connectionLock, boolean asnycConnection) {
        this.connectionLock = connectionLock;
        this.asnycConnection = asnycConnection;
    }

    @Override
    public void processResult(int rc, String path, Object ctx, List<String> children) {
    }

    @Override
    public void process(WatchedEvent event) {
        if (!asnycConnection && event.getState() == Event.KeeperState.SyncConnected) {
            connectionLock.release();//连接成功
        } else if (event.getState() == Event.KeeperState.Disconnected) {//连接断开
            log.warn("zookeeper 连接断开......");
        } else if (event.getState() == Event.KeeperState.Expired) {//连接超时
            log.warn("zookeeper session timeout......");
            try {
                zkClient.reconnect();
                process.relisten();
                process.recreate();
            } catch (Exception e) {
                log.error("reconnection to zookeeper error.", e);
            }
        }

        if (event.getType() == Event.EventType.NodeChildrenChanged) {//子节点变化
            process.clientNodeChanged(event.getPath());
        } else if (event.getType() == Event.EventType.NodeDataChanged) {//节点数据变化
            process.nodeDataChanged(event.getPath());
        }
    }

    public void setWatcherProcess(WatcherProcess process, ZkClient zkClient) {
        this.process = process;
        this.zkClient = zkClient;
    }

    public WatcherProcess getProcess() {
        return process;
    }
}
